package com.atguigu.Flink.sql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink04_UpsertKafkaConnector {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String createSourceTable =
                "CREATE TABLE t1 (\n" +
                        "  `id` STRING,\n" +
                        "  `vc` INT,\n" +
                        "  `ts` BIGINT\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'topicA',\n" +
                        "  'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092',\n" +
                        "  'properties.group.id' = 'flink12',\n" +
                        "  'scan.startup.mode' = 'group-offsets',\n" +
                        "  'properties.auto.offset.reset' = 'latest' ," +
                        "  'format' = 'csv'\n" +
                        ")" ;
        tableEnv.executeSql(createSourceTable);

        Table newtable = tableEnv.sqlQuery("select id,sum(vc) svc from t1 group by id");
        tableEnv.createTemporaryView("new_table",newtable);

        //kafka connector write
        String createSinkTable =
                "CREATE TABLE t2 (\n" +
                        " id STRING ," +
                        " svc BIGINT , " +
                        " PRIMARY KEY (id) NOT ENFORCED " +
                        ") WITH (\n" +
                        "  'connector' = 'upsert-kafka',\n" +
                        "  'topic' = 'topicB',\n" +
                        "  'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092',\n" +
                        "  'key.format' = 'csv' , \n" +
                        "  'value.format' = 'csv' " +
                        ")" ;

        tableEnv.executeSql(createSinkTable);

        tableEnv.executeSql("insert into t2 select id,svc from new_table");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
